package com.pw.study.flink.transfrom

import com.pw.study.flink.entities.UserBean
import com.pw.study.flink.source.CustomSource
import org.apache.flink.streaming.api.scala._

import scala.collection.mutable.ArrayBuffer

object FlatMapTF {
  def main(args: Array[String]): Unit = {
    //GT_1204pw@010
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val ds: DataStream[UserBean] = env.addSource(new CustomSource())

    val dsMap: DataStream[ArrayBuffer[UserBean]] = ds.map(new MyRichMap())
    val dsFlagMap: DataStream[UserBean] = dsMap.flatMap(new MyFlatMap())

    dsFlagMap.print("FlatMap: ")
    ds.print("ds: ")


    env.execute("jobName")
  }
}
